package com.kafka0102.demo;

import java.util.logging.Level;
import java.util.logging.Logger;

import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandler.Sharable;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.handler.codec.oneone.OneToOneEncoder;

@Sharable
public class MessageEncoder extends OneToOneEncoder {
	private static final Logger logger = Logger.getLogger(MessageEncoder.class.getName());

	@Override
	protected Object encode(ChannelHandlerContext ctx, Channel channel, Object msg) throws Exception {
		logger.log(Level.INFO, "message encoder encode.");
		if (!(msg instanceof String)) {
			return msg;//(1)
		}

		String res = (String) msg;
		byte[] data = res.getBytes();
		int dataLength = data.length;
		ChannelBuffer buf = ChannelBuffers.dynamicBuffer();//(2)
		buf.writeInt(dataLength);
		buf.writeBytes(data);
		return buf;//(3)
	}
}
